﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;

namespace Microsoft.Media.IngestClient
{
    class ByteQueue
    {
        int count = 0;
        int capacity = 1024;
        byte[] data;
        int tail = 0;
        int head = 0;
        double growthFactor = .1; // 10%

        public ByteQueue(int capacity)
        {
            this.capacity = capacity;
            data = new byte[this.capacity];
        }

        public ByteQueue()
        {
            this.capacity = 32;
            data = new byte[this.capacity];
        }

        public void EnqueueMany(byte[] bytes, int offset, int length)
        {
            int neededCapacity = length + count;
            if (capacity < neededCapacity) // grow data array
            {
                // copy the current state
                byte[] oldData = data;
                int oldHead = head;
                int oldTail = tail;
                int oldCapcity = capacity;
                // reset the new queue
                capacity = (int)((1.0 + growthFactor) * neededCapacity);
                data = new byte[capacity];
                head = 0;
                tail = count;
                // copy the data over
                if (oldTail > oldHead)
                {
                    System.Buffer.BlockCopy(oldData, oldHead, data, 0, count);
                }
                else
                {
                    int bytesToCopy = oldCapcity - oldHead;
                    System.Buffer.BlockCopy(oldData, oldHead, data, 0, bytesToCopy);
                    System.Buffer.BlockCopy(oldData, 0, data, bytesToCopy, count - bytesToCopy);
                }
            }

            int roomAfterTail = head <= tail ? capacity - tail : head - tail;
            if (roomAfterTail >= length) // we can do it in one shot
            {
                System.Buffer.BlockCopy(bytes, offset, data, tail, length);
                tail += length;
                count += length;
            }
            else // it will require 2 copy operations
            {
                System.Buffer.BlockCopy(bytes, offset, data, tail, roomAfterTail);
                int bytesLeftToCopy = length - roomAfterTail;
                System.Buffer.BlockCopy(bytes, offset + roomAfterTail, data, 0, bytesLeftToCopy);
                tail = bytesLeftToCopy;
                count += length;
            }
        }

        public void DequeueMany(byte[] bytes, int offset, int length)
        {
            int roomAfterHead = tail > head ? tail - head : capacity - head;
            if (roomAfterHead >= length) // we can do it in one shot
            {
                System.Buffer.BlockCopy(data, head, bytes, offset, length);
                head += length;
                count -= length;
            }
            else // it will require 2 copy operations
            {
                System.Buffer.BlockCopy(data, head, bytes, offset, roomAfterHead);
                int bytesLeftToCopy = length - roomAfterHead;
                System.Buffer.BlockCopy(data, 0, bytes, offset + roomAfterHead, bytesLeftToCopy);
                head = bytesLeftToCopy;
                count -= length;
            }
        }

        public int Count { get { return count; } }
    }

    class IOStreamManager : IDisposable
    {
        readonly ByteQueue sharedBuffer;
        readonly InputStream inputStream;
        readonly OutputStream outputStream;

        public IOStreamManager()
        {
            sharedBuffer = new ByteQueue(1024 * 1024);
            inputStream = new InputStream(sharedBuffer);
            outputStream = new OutputStream(sharedBuffer);
        }

        public Stream InputStream { get { return inputStream; } }
        public Stream OutputStream { get { return outputStream; } }

        /// <summary>
        /// Call to indicate that the OutputStream has been fully written to and flushed. This will tell the InputStream to no longer continue to wait for bytes.
        /// </summary>
        public void SignalComplete()
        {
            inputStream.SetEndOfStream(OutputStream.Length);
        }

        public void Dispose()
        {
            OutputStream.Dispose();
            InputStream.Dispose();
        }
    }

    class OutputStream : Stream
    {
        const int BufferLength = 1024 * 64; // this happens to be the chunk size that the HttpClient reads during a POST. No need to go slower or faster than it.

        long position;
        ByteQueue sharedBuffer;
        byte[] buffer;
        int bufferPosition;

        internal OutputStream(ByteQueue sharedBuffer)
        {
            this.sharedBuffer = sharedBuffer;
            buffer = new byte[BufferLength];
        }

        public override bool CanRead
        {
            get { return false; }
        }

        public override bool CanSeek
        {
            get { return false; }
        }

        public override bool CanWrite
        {
            get { return true; }
        }

        public override void Flush()
        {
            lock (sharedBuffer)
            {
                sharedBuffer.EnqueueMany(buffer, 0, bufferPosition);
                position += bufferPosition;
                bufferPosition = 0; // reset buffer position

                Monitor.Pulse(sharedBuffer);
            }
        }

        public override long Length
        {
            get { return position; }
        }

        public override long Position
        {
            get
            {
                return position;
            }
            set
            {
                throw new NotImplementedException();
            }
        }

        public override int Read(byte[] buffer, int offset, int count)
        {
            throw new NotImplementedException();
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            throw new NotImplementedException();
        }

        public override void SetLength(long value)
        {
            throw new NotImplementedException();
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            while (count > 0)
            {
                var bytesToWrite = Math.Min(BufferLength - bufferPosition, count);

                System.Buffer.BlockCopy(buffer, offset, this.buffer, bufferPosition, bytesToWrite);

                bufferPosition += bytesToWrite;
                count -= bytesToWrite;
                offset += bytesToWrite;

                if (bufferPosition == BufferLength) // we filled the internal buffer, flush data
                {
                    this.Flush();
                }
            }
        }
    }

    class InputStream : Stream
    {
        bool isDisposed;
        long position;
        ByteQueue sharedBuffer;
        long? maxLength;

        internal InputStream(ByteQueue sharedBuffer)
        {
            this.sharedBuffer = sharedBuffer;
        }

        public override bool CanRead
        {
            get { return true; }
        }

        public override bool CanSeek
        {
            get { return false; }
        }

        public override bool CanWrite
        {
            get { return false; }
        }

        public override void Flush()
        {
            throw new NotImplementedException();
        }

        public override long Length
        {
            get { return position + sharedBuffer.Count; }
        }

        public override long Position
        {
            get
            {
                return position;
            }
            set
            {
                throw new NotImplementedException();
            }
        }

        public override int Read(byte[] buffer, int offset, int count)
        {
            if (maxLength.HasValue && position == maxLength.Value) return 0; // indicate end of stream

            int bytesToRead;

            lock (sharedBuffer)
            {
                while ((bytesToRead = Math.Min(count, sharedBuffer.Count)) == 0)
                {
                    Monitor.Wait(sharedBuffer);
                    if (isDisposed) return 0;
                    if (maxLength.HasValue && position + bytesToRead == maxLength.Value) break; // we've read everything we can, return with what we have.
                }

                sharedBuffer.DequeueMany(buffer, offset, bytesToRead);
                position += bytesToRead;
                Monitor.Pulse(sharedBuffer);
            }
            return bytesToRead;
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            throw new NotImplementedException();
        }

        public override void SetLength(long value)
        {
            throw new NotImplementedException();
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            throw new NotImplementedException();
        }

        public void SetEndOfStream(long position)
        {
            maxLength = position;
            lock (sharedBuffer)
                Monitor.Pulse(sharedBuffer);
        }

        protected override void Dispose(bool disposing)
        {
            isDisposed = true;
            lock (sharedBuffer)
                Monitor.Pulse(sharedBuffer);
            base.Dispose(disposing);
        }
    }
}
